package com.atguigu.day03;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.socketTextStream("192.168.1.102", 9999);
        streamSource.keyBy(x->true).process(new KeyedProcessFunction<Boolean, String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                long currentTime = ctx.timerService().currentProcessingTime();
                out.collect("元素: "+ value + "开始计算，当前时间为"+new Timestamp(currentTime));
                long onTimer = currentTime + 20 * 1000;
                ctx.timerService().registerProcessingTimeTimer(onTimer);
                out.collect("注册了定时器，触发时间为："+new Timestamp(onTimer));
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                out.collect("定时器触发  ，触发时间为："+new Timestamp(timestamp));
            }
        }).print();
        env.execute();
    }
}
